//rcclust-master.c
//Programmer: Rory Carmichael
//Date: 1/21/2011
//Purpose: C implementation of pvaluelooper.pl
//UPDATE: Adding Master-Worker functionality as of 2/17/2011
//UPDATE: This is desperately in need of some sort of error handling with regards to failed work-queue jobs 3/8/2011

#include "work_queue.h"

#include <stdio.h>
#include <unistd.h>
#include <ctype.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <math.h>
#include <errno.h>

//This could use some cleanup and refactoring
float* calc_peak_frequencies(char *filename, int *numcols) {
	int colsize = 0;
	int i = 0;
	int j = 0;
	int first = 1;
	float *freqs;
	int *lengths;
	int *peaks;
	FILE *infile;
	char line[ 10000 ];
	char line2[ 10000 ];
	char delims[] = "\t\n";
	char *result = NULL;
	colsize = *numcols;
	infile = fopen(filename, "rt");
	if ( infile != NULL ) {
		while ( fgets(line, sizeof(line), infile) != NULL) {
			if(first && line != NULL) {
				fgets(line, sizeof(line), infile);
				fgets(line, sizeof(line), infile);
				first = 0;
				memcpy(line2,line,sizeof(char)*10000);
				result = strtok(line, delims);
				while ( result != NULL && (strcmp(result, "\n") != 0)) {
					colsize++;
					result = strtok(NULL, delims);
				}
				lengths = (int *) malloc(colsize*sizeof(int));
				peaks = (int *) malloc(colsize*sizeof(int));
				freqs = (float *) malloc(colsize*sizeof(float));
				for(i=0; i<colsize; i++) {
					lengths[i] = 0;
					peaks[i] = 0;
					freqs[i] = 0.0;
				}
				result = strtok(line2, delims);
				i = 0;
				while ( result != NULL && (strcmp(result, "\n") != 0)) {
					if (strcmp(result, "-") != 0) {
						lengths[i]++;
						if (strcmp(result, "0") == 0) {
							peaks[i]++;
						}
					}
					i++;
					result = strtok(NULL, delims);
				}
			} else {
				result = strtok(line, delims);
				i = 0;
				while ( result != NULL && (strcmp(result, "\n") != 0)) {
					if (strcmp(result, "-") != 0) {
						lengths[i]++;
						if (strcmp(result, "0") == 0) {
							peaks[i]++;
						}
					}
					i++;
					result = strtok(NULL, delims);
				}
			}
		}
		fclose(infile);
		for(i=0;i<colsize;i++) { 
			freqs[i] = ((float) peaks[i]) / ((float) lengths[i]);
		}
		*numcols = colsize;
		return freqs;
	} else {
		printf("Error while opening %s\n", filename);
		perror(filename);
	}
	return NULL;
}

int submit_file_job(char *filename, struct work_queue *q) {
	struct work_queue_task *t;
	FILE *infile;
	float *pvallist = NULL;
	char *result = NULL;
	char delims[] = "\t\n";
	char line [ 10000 ];
	char filebuffer [ 250000 ];
	int *distlist = NULL;
	int numcols = 0;
	int splitsize = 25;
	int splitcount = 0;
	int splitnum = 0;
	char command[256], outfile[256], pvalname[256];
	FILE *pvalfile;
	FILE *tmpfile;

	sprintf(pvalname,"%s.pvalfile",filename);
	pvallist = calc_peak_frequencies(filename, &numcols);
	pvalfile = fopen(pvalname, "wb");
	fwrite(&numcols,sizeof(int),sizeof(int)/sizeof(int),pvalfile);
	fwrite(pvallist,sizeof(float),numcols,pvalfile);
	fclose(pvalfile);

	distlist = (int *) malloc(numcols*sizeof(int));
	if(pvallist != NULL) {
		infile = fopen(filename, "rt");
		if(infile == NULL) {
			fprintf(stderr, "Failed to open %s\n", filename);
			exit(1);
		}
		fgets(line,sizeof(line),infile);

		if (line != NULL) {
			result = strtok(line, delims);
			sprintf(outfile,"%s.%d.out",filename,splitnum);
			tmpfile = fopen(outfile,"w");
			//fprintf(tmpfile, "%s\n\n", line);
			fprintf(tmpfile, "%s\n\n", result);
			fclose(tmpfile);
			splitnum++;
		} else {
			return 1;
		}
		fgets(line,sizeof(line),infile);
		sprintf(filebuffer,"");
		while (fgets(line,sizeof(line),infile) != NULL && strcmp(line, "\n") != 0) {
			sprintf(filebuffer,"%s%s",filebuffer,line);
			splitcount++;
			if(splitcount >= splitsize) {
				sprintf(outfile,"%s.%d",filename,splitnum);
				sprintf(command,"./rc-calc-row-pvalue pvalfile linefile");
				t = work_queue_task_create(command);
				work_queue_task_specify_input_file(t,"rc-calc-row-pvalue","rc-calc-row-pvalue");
				work_queue_task_specify_input_buf(t,filebuffer,sizeof(line)*splitcount,"linefile");
				work_queue_task_specify_input_file(t,pvalname,"pvalfile");
				work_queue_task_specify_tag(t,outfile);
				work_queue_submit(q,t);

				splitcount=0;
				splitnum++;
				sprintf(filebuffer,"");
			}
		}
		sprintf(outfile,"%s.%d",filename,splitnum);
		sprintf(command,"./rc-calc-row-pvalue pvalfile linefile");
		t = work_queue_task_create(command);
		work_queue_task_specify_input_file(t,"rc-calc-row-pvalue","rc-calc-row-pvalue");
		work_queue_task_specify_input_buf(t,filebuffer,sizeof(line)*splitcount,"linefile");
		work_queue_task_specify_input_file(t,pvalname,"pvalfile");
		work_queue_task_specify_tag(t,outfile);
		work_queue_submit(q,t);
		fclose(infile);
	}
	return 0;
}

int print_joined_output(char * filename, int numfiles) {
	char outfile[256];
	char tmpname[256];
	char pvalname[256];
	FILE *tmpfile;
	int i=0;
	char line[ 10000 ];

	if(numfiles == 1 && file_exists(filename)) {
		sprintf(tmpname,"%s",filename);
	} else {
		sprintf(tmpname,"%s.%d",filename,i);
	} 
	while(i < numfiles && file_exists(tmpname)) {
		fprintf(stderr, "printing output from %s jobs...\n",tmpname);
		int j = 0;
		sprintf(outfile,"%s.%d.out",tmpname,j);
		while(file_exists(outfile)) {
			tmpfile = fopen(outfile,"r");
			if(tmpfile != NULL) {
				while(fgets(line,sizeof(line),tmpfile) != NULL ) {
					fprintf(stdout,"%s",line);
				}
				fclose(tmpfile);
				remove(outfile);
			} else {
				fprintf(stderr,"ERROR: could not open %s\n",outfile);
				return 1;
			}
			j++;
			sprintf(outfile,"%s.%d.out",tmpname,j);
		}
		i++;
		sprintf(pvalname,"%s.pvalfile",tmpname);
		remove(pvalname);
		fprintf(stdout,"\n");
		sprintf(tmpname,"%s.%d",filename,i);
	}
	return 0;
}

int file_exists(char * filename) {
	FILE *testfile;
	testfile = fopen(filename, "r");
	if(testfile) {
		fclose(testfile);
		return 1;
	} else {
		return 0;
	}
}

int collect_output(struct work_queue *q) {
	struct work_queue_task *t;

	while(!work_queue_empty(q)) {
		t= work_queue_wait(q,5);
		if(t) {
			task_complete(t);
		}
	}
	return 0;
}

int task_complete(struct work_queue_task *t) {
	char outfile[256];
	FILE *tmpfile;

	sprintf(outfile,"%s.out",t->tag);
	fprintf(stderr,"received output from %s\n",outfile);
	tmpfile = fopen(outfile, "w");
	fprintf(tmpfile,"%s",t->output);
	fclose(tmpfile);
	work_queue_task_delete(t);
	
	return 0;
}

int main (int argc, char **argv) {
	struct work_queue *q;
	struct work_queue_task *task;
	int port = WORK_QUEUE_DEFAULT_PORT;
	FILE *infile;
	FILE *tmpfile;
	int i = 0;
	int j = 0;
	int numfiles = 0;
	int index = 0;
	float *pvallist = NULL;
	char *filename = NULL;
	char *result = NULL;
	char delims[] = "\t\n";
	char line [ 10000 ];
	char filebuffer [ 250000 ];
	int *distlist = NULL;
	int numcols = 0;
	int splitsize = 50;
	int splitcount = 0;
	char command[256], outfile[256];
	FILE *pvalfile;
	char tmpname[256];

	if (argc - index != 3) {
		fprintf(stderr, "USAGE: %s <prefix> <# files>\n", argv[0]);
		exit(1);
	}
	index++;

	q = work_queue_create(port);
	if(!q) {
		printf("couldn't listen on port %d: %s\n",port,strerror(errno));
		return 1;
	}
	fprintf(stderr,"listening on port %d...\n",port);

	filename = argv[index];
	index++;
	sscanf(argv[index],"%d",&numfiles);
	i=0;
	while(1) {
		while((i < numfiles) && work_queue_hungry(q)) {
			if(numfiles == 1 && file_exists(filename)) {
				sprintf(tmpname,"%s",filename);
			} else {
				sprintf(tmpname,"%s.%d",filename,i);
			}
			if(file_exists(tmpname)) {
				fprintf(stderr,"submitting jobs for file: %s...\n",tmpname);
				submit_file_job(tmpname,q);
				i++;
			} else {
				fprintf(stderr,"Didn't see a file with name: %s\n",tmpname);
			}
		}

		if ((i >= numfiles) && work_queue_empty(q)) break;

		task = work_queue_wait(q,5);
		if(task) {
			if(task->result == WORK_QUEUE_RESULT_FUNCTION_FAIL) {
				fprintf(stderr,"Resubmitting failed task: %s",task->tag); 
				work_queue_submit(q,task);
			} else {
				task_complete(task);
			}
		}
	}
	fprintf(stderr,"printing collected output files from submissions\n");
	print_joined_output(filename, numfiles);

	work_queue_delete(q);
	return 0;
}
